package com.bw.gmall.realtime.lyt;

import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.HeartbeatManagerOptions;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

public class DwdTradeOrderRefund {
    public static void main(String[] args) throws Exception {

        // 替换原来的env初始化代码
        Configuration conf = new Configuration();
        // 心跳间隔设为5秒，超时时间设为60秒（根据集群情况调整）
        conf.setLong(HeartbeatManagerOptions.HEARTBEAT_INTERVAL, 5000);
// 增加更多Flink配置参数，提高稳定性
        conf.setLong(HeartbeatManagerOptions.HEARTBEAT_TIMEOUT, 120000); // 延长至120秒

// 本地调试时启用WebUI，方便观察作业状态
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //读取kafka主题中的数据库变更数据
        tableEnv.executeSql(MyKafkaUtil.getTopicDb("topic_db"));



        //读取退单表数据
        Table orderRefundInfo = tableEnv.sqlQuery(" select\n" +
                "data['id'] id,\n" +
                "data['user_id'] user_id,\n" +
                "data['order_id'] order_id,\n" +
                "data['sku_id'] sku_id,\n" +
                "data['refund_type'] refund_type,\n" +
                "data['refund_num'] refund_num,\n" +
                "data['refund_amount'] refund_amount,\n" +
                "data['refund_reason_type'] refund_reason_type,\n" +
                "data['refund_reason_txt'] refund_reason_txt,\n" +
                "data['create_time'] create_time,\n" +
                "pt \n" +
                "from topic_db\n" +
                "where `table` = 'order_refund_info'\n" +
                "and `type` = 'insert'\n"
        );
        tableEnv.createTemporaryView("order_refund_info",orderRefundInfo);
        // tableEnv.executeSql("select * from order_refund_info").print();
        System.out.println("order_refund_info数据量：");
        TableResult refundInfoResult = tableEnv.executeSql("SELECT COUNT(*) FROM order_refund_info");
        refundInfoResult.print();

        // 读取新的字典表数据，用于后续维表关联
        tableEnv.executeSql(getBaseDicLookUpDDL());
        // 调试输出：检查数据
        try {
            System.out.println("字典表前10条:");
            tableEnv.executeSql("SELECT * FROM  base_dic_new LIMIT 10").print();
        } catch (Exception e) {
            System.err.println("查询字典表关联数据失败: " + e.getMessage());
        }

        //4. 提取订单信息表中的退款相关数据
        //条件：订单状态变更为"1005"(退款成功)且状态发生了变化
        Table orderInfoRefund = tableEnv.sqlQuery("select\n" +
                "data['id'] id,\n" +
                "data['province_id'] province_id,\n" +
                "`old`\n" +
                "from topic_db\n" +
                "where `table` = 'order_info'\n" +
                "and `type` = 'update'\n" +
                "and data['order_status']='1005'\n" +
                "and `old`['order_status'] is not null"
        );
        tableEnv.createTemporaryView("order_info_refund", orderInfoRefund);
//        tableEnv.executeSql("select * from order_info_refund").print();


        //读取详情表
        //读取订单数据 条件筛选退单数据
        Table orderDetailRefund = tableEnv.sqlQuery("select\n" +
                "data['order_id'] order_id,\n" +
                "data['sku_id'] sku_id\n" +
                "from topic_db\n" +
                "where `table` = 'order_detail'\n"
                + "and `type` = 'insert'\n"
        );
        tableEnv.createTemporaryView("order_detail", orderDetailRefund);
//        tableEnv.executeSql("select * from order_detail").print();

        //提取商品店铺关联表数据
        tableEnv.executeSql(getShopskuDDL());
        // 调试输出：检查数据
        try {
            System.out.println("商品表前10条:");
            tableEnv.executeSql("SELECT * FROM  shop_sku LIMIT 10").print();
        } catch (Exception e) {
            System.err.println("查询商品表表关联数据失败: " + e.getMessage());
        }

        // 构建宽表（关联所有表，新增 order_detail 和 shop_sku）
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "ri.id,\n" +
                "ri.user_id,\n" +
                "ri.order_id,\n" +
                "ri.sku_id,\n" +
                "oi.province_id,\n" +
                "CAST(ss.shop_id AS BIGINT) AS shop_id, \n" +
                "date_format(ri.create_time, 'yyyy-MM-dd') date_id,\n" +
                "ri.create_time,\n" +
                "ri.refund_type as refund_type_code,\n" +  // 明确字段别名
                "type_dic.dic_name as refund_type_name,\n" +
                "ri.refund_reason_type as refund_reason_type_code,\n" +
                "reason_dic.dic_name as refund_reason_type_name,\n" +
                "ri.refund_reason_txt,\n" +
                "ri.refund_num,\n" +
                "ri.refund_amount,\n" +
                "ri.pt,\n" +
                "current_row_timestamp() row_op_ts\n" +  //处理时间
                "from order_refund_info ri\n" +
                // 关联订单表（获取省份ID）
                "join order_info_refund oi\n" +
                "on CAST(ri.order_id AS STRING) = CAST(oi.id AS STRING) \n" +
                // 关联订单详情表（验证订单与商品关联）
                "join order_detail od\n" +
                "on TRIM(ri.order_id) = TRIM(od.order_id) and TRIM(ri.sku_id) = TRIM(od.sku_id) \n" +
                // 关联商品-店铺表（获取店铺ID）
                " join shop_sku ss\n" +
                " on CAST(ri.sku_id AS BIGINT) = ss.sku_id\n" +
                "   AND ss.shop_id IS NOT NULL \n" +  // 通过 sku_id 关联店铺
                // 关联字典表（退款类型）
                " left join base_dic_new for system_time as of ri.pt as type_dic\n" +
                "on CAST(ri.refund_type AS STRING) = CAST(type_dic.base_id AS STRING)\n" +
                // 关联字典表（退款原因）
                " left join base_dic_new for system_time as of ri.pt as reason_dic\n" +
                "on CAST(ri.refund_reason_type AS STRING) = CAST(reason_dic.base_id AS STRING) "
        );
        tableEnv.createTemporaryView("result_table", resultTable);
//         tableEnv.executeSql("select * from result_table").print();
        System.out.println("result_table数据量：");
        TableResult resultResult = tableEnv.executeSql("SELECT COUNT(*) FROM result_table");
        resultResult.print();

        // 7. 创建 DWD 层输出表（新增 shop_id 字段）
        tableEnv.sqlQuery("create table dwd_trade_order_refund(\n" +
                "id string,\n" +
                "user_id string,\n" +
                "order_id string,\n" +
                "sku_id string,\n" +
                "province_id string,\n" +
                "shop_id bigint, \n" +
                "date_id string,\n" +
                "create_time string,\n" +
                "refund_type_code string,\n" +
                "refund_type_name string,\n" +
                "refund_reason_type_code string,\n" +
                "refund_reason_type_name string,\n" +
                "refund_reason_txt string,\n" +
                "refund_num string,\n" +
                "refund_amount string,\n" +
                "pt timestamp_ltz(3) ,\n" +
                "row_op_ts timestamp_ltz(3)\n" +
                ")" + MyKafkaUtil.getKafkaSinkDDL("dwd_trade_order_refund_a"));



        //将处理结果写入kafka
        tableEnv.executeSql(" " +
                "insert into dwd_trade_order_refund select * from result_table");
//        //打印输出结果
//        TableResult tableResult = tableEnv.executeSql("select * from dwd_trade_order_refund");
//        tableResult.print();


        env.execute("DwdTradeOrderRefundJob");


    }
    //-------------------------------------------------------------------------------------------------------------------------------
    //字典表
    public static String getBaseDicLookUpDDL() {
        return "create table `base_dic_new`( " +
                "`base_id` string, " + // 字典编码（如退款类型编码：1、2、3）
                "`dic_name` string, " + // 编码对应的名称（如“退货退款”“仅退款”）
                "`parent_code` string, " + //父级编号
                "primary key(`base_id`) not enforced " +  // 声明主键（Flink不强制校验，仅用于优化）
                ")" + mysqlLookUpTableDDL("base_dic_new"); // 拼接JDBC连接参数
    }

    public static String mysqlLookUpTableDDL(String tableName) {

        return " WITH ( " +
                "'connector' = 'jdbc', " + // 连接器类型：JDBC（用于连接MySQL）
                "'url' = 'jdbc:mysql://hadoop102:3306/gmall', " +  // MySQL连接地址（hadoop102为主机名，3306为端口，gmall为数据库名）
                "'table-name' = '" + tableName + "', " + // 关联的MySQL表名（这里传入的是base_dic）
                "'lookup.cache.max-rows' = '1000', " + // 维表缓存的最大行数（缓存最近10条查询结果）
                "'lookup.cache.ttl' = '1 hour', " + // 缓存过期时间（1小时内重复查询相同key，直接用缓存）
                "'username' = 'root', " +  // MySQL登录用户名
                "'password' = 'root', " + // MySQL登录密码
                "'driver' = 'com.mysql.cj.jdbc.Driver' " + // MySQL驱动类
                ")";
    }


    //商品shop表
    public static String getShopskuDDL() {
        return "create table `shop_sku`( " +
                "`sku_id` bigint, " +
                "`shop_id` BIGINT, " +
                "`create_time` TIMESTAMP(6) , " +
                "primary key(`sku_id`) not enforced " +  // 声明主键（Flink不强制校验，仅用于优化）
                ")" + mysqlShopskuDDL("shop_sku"); // 拼接JDBC连接参数
    }

    public static String mysqlShopskuDDL(String tableName) {

        return " WITH ( " +
                "'connector' = 'jdbc', " + // 连接器类型：JDBC（用于连接MySQL）
                "'url' = 'jdbc:mysql://hadoop102:3306/gmall', " +  // MySQL连接地址（hadoop102为主机名，3306为端口，gmall为数据库名）
                "'table-name' = '" + tableName + "', " + // 关联的MySQL表名（这里传入的是base_dic）
                "'lookup.cache.max-rows' = '100', " + // 维表缓存的最大行数（缓存最近10条查询结果）
                "'lookup.cache.ttl' = '1 hour', " + // 缓存过期时间（1小时内重复查询相同key，直接用缓存）
                "'username' = 'root', " +  // MySQL登录用户名
                "'password' = 'root', " + // MySQL登录密码
                "'driver' = 'com.mysql.cj.jdbc.Driver' " + // MySQL驱动类
                ")";
    }


}
